// com/tidu/template/impl/DisruptorDataSyncTemplate.java
package com.tidu.template.impl;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.tidu.disruptor.event.DataSyncEvent;
import com.tidu.disruptor.factory.DataSyncEventFactory;
import com.tidu.disruptor.handler.WriteDataEventHandler;
import com.tidu.disruptor.manager.BatchAckManager;
import com.tidu.disruptor.product.DataSyncEventProducer;
import com.tidu.strategy.SingleTableSyncDisruptorStrategy;
import com.tidu.strategy.sync.SyncStrategy;
import com.tidu.template.AbstractDataSyncTemplate;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class DisruptorDataSyncTemplate extends AbstractDataSyncTemplate {
    private Disruptor<DataSyncEvent> disruptor;
    private RingBuffer<DataSyncEvent> ringBuffer;
    private DataSyncEventProducer producer;
    private volatile boolean isShutdown = false;
    private final ExecutorService executorService = Executors.newFixedThreadPool(1);
    private final AtomicBoolean isRunning = new AtomicBoolean(false);

    // 批次确认管理器
    private final BatchAckManager batchAckManager;

    // 批次ID生成器
    private final AtomicLong batchIdGenerator = new AtomicLong(0);

    public DisruptorDataSyncTemplate(SyncStrategy syncStrategy) {
        super(syncStrategy);
        this.batchAckManager = new BatchAckManager(syncStrategy);
        initDisruptor();
    }

    private void initDisruptor() {
        // 创建 Disruptor 实例
        disruptor = new Disruptor<>(
                new DataSyncEventFactory(),
                16,
                Executors.defaultThreadFactory(),
                ProducerType.SINGLE,
                new BlockingWaitStrategy() // 使用BlockingWaitStrategy节省CPU
        );

        // 创建事件处理器，传入批次确认管理器
        WriteDataEventHandler writeHandler1 = new WriteDataEventHandler(syncStrategy, batchAckManager, "handler-1");
        WriteDataEventHandler writeHandler2 = new WriteDataEventHandler(syncStrategy, batchAckManager, "handler-2");
        WriteDataEventHandler writeHandler3 = new WriteDataEventHandler(syncStrategy, batchAckManager, "handler-3");
        WriteDataEventHandler writeHandler4 = new WriteDataEventHandler(syncStrategy, batchAckManager, "handler-4");
//        ErrorHandler errorHandler = new ErrorHandler(syncStrategy, batchAckManager, "errorHandler");

        // 设置处理器
        disruptor.handleEventsWithWorkerPool(
                writeHandler1,
                writeHandler2,
                writeHandler3,
                writeHandler4
        );

        // 启动 Disruptor
        ringBuffer = disruptor.start();

        // 创建生产者
        producer = new DataSyncEventProducer(ringBuffer);
    }

    @Override
    public void execute() {
        isRunning.set(true);
        try {
            long startTime = System.currentTimeMillis();

            // 使用批次确认管理器的最后确认offset来获取下一批数据
            long offset = batchAckManager.getNextOffset();

            while (!syncStrategy.isCompleted() && isRunning.get()) {
                List<Map<String, Object>> data = syncStrategy.fetchData(offset);

                if (data != null && !data.isEmpty()) {
                    if(ringBuffer.remainingCapacity() < 10){
                        System.out.println("================================>*************************小于10了*********************<===============================");
                    }
                    if (ringBuffer.remainingCapacity() < 5 && isRunning.get()){
                        Thread.sleep(2000);
                    } else if (ringBuffer.remainingCapacity() < 10 && isRunning.get()){
                        Thread.sleep(500);
                    }

                    // 创建带批次信息的事件
                    DataSyncEvent batchEvent = new DataSyncEvent();
                    batchEvent.setData(data);
                    batchEvent.setBatchId(batchIdGenerator.incrementAndGet()); // 批次ID,从1开始
                    batchEvent.setOffset(offset); // 设置当前批次的起始offset
                    batchEvent.setBatchSize(data.size());
                    //写一批数据的offset
                    offset += data.size();

                    // 注册批次到确认管理器
                    batchAckManager.registerBatch(batchEvent);

                    // 发布数据事件
                    producer.onData(batchEvent);
                } else {
                    // 没有更多数据
                    break;
                }
            }

            // 等待所有事件处理完成
            waitForProcessingToComplete();

            long endTime = System.currentTimeMillis();
            log.info("Data sync completed in {} s", (endTime - startTime) / 1000.0);
            log.info("Remaining pending batches: {}", batchAckManager.getPendingBatchCount());
            log.info("Failed batches: {}", batchAckManager.getFailedBatchCount());

        } catch (Exception e) {
            log.error("Error during data sync execution", e);
            throw new RuntimeException("Error during data sync execution", e);
        } finally {
            isRunning.set(false);
            shutdown();
        }
    }

    /**
     * 等待所有事件处理完成
     */
    private void waitForProcessingToComplete() {
        try {
            log.info("Waiting for all events to be processed...");
            // 等待所有待处理批次完成
            long startTime = System.currentTimeMillis();
            while (batchAckManager.getPendingBatchCount() > 0 &&
                    (System.currentTimeMillis() - startTime) < 120000) { // 最多等待2分钟
                log.info("Waiting for {} pending batches to complete...", batchAckManager.getPendingBatchCount());
                Thread.sleep(1000);
            }

            if (batchAckManager.getPendingBatchCount() > 0) {
                log.warn("Some batches are still pending after timeout");
            }

            disruptor.shutdown(120, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.warn("Disruptor did not shutdown gracefully", e);
            disruptor.shutdown();
        }
    }

    /**
     * 关闭资源
     */
    private void shutdown() {
        if (!isShutdown) {
            isRunning.set(false);
            try {
                executorService.shutdown();
                if (!executorService.awaitTermination(15, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }

            try {
                // 如果是SingleTableSyncDisruptorStrategy，关闭其资源
                if (syncStrategy instanceof SingleTableSyncDisruptorStrategy) {
                    ((SingleTableSyncDisruptorStrategy) syncStrategy).shutdown();
                }

                if (disruptor != null) {
                    disruptor.shutdown();
                }
                log.info("Disruptor has been shut down");
            } catch (Exception e) {
                log.error("Error shutting down Disruptor", e);
            } finally {
                isShutdown = true;
            }
        }
    }

    /**
     * 停止同步过程
     */
    public void stop() {
        isRunning.set(false);
        log.info("Stopping data sync process...");
    }
}
